package org.anno.chunkdb;

import it.unimi.dsi.fastutil.ints.AbstractIntIterator;
import it.unimi.dsi.fastutil.ints.IntIterator;
import it.unimi.dsi.fastutil.ints.IntIterators;

import org.anno.chunkdb.Metadata.Column.Type;

import java.io.IOException;
import java.util.Iterator;

/**
 * Superficial decoding of a {@link Chunkdb.Chunk}. Supports {@code Host<RowVisitor>} with lazy
 * decoding.
 *
 * <pre>
 * Value {
 *   varInt length;
 *   byte[length] data;
 * }
 * Column {
 *   Value[row_count] value;
 *   int20[row_count] value_start;
 *   int16[row_count] index;  // value[index[i]] <= value[index[j]] if i <= j
 * }
 * Chunk {
 *   int16 version;
 *   Column[column_count] column;
 *   int24[column_count] column_start;
 *   int16 row_count;
 * }
 * </pre>
 */
public class LightChunk implements QueryPlanSupport {

  private final byte[] flatArray;
  private final ColumnQuerySupport[] column;
  private final int rowCount;

  public LightChunk(final byte[] flatArray, Metadata.Table table) {
    this.flatArray = flatArray;
    int ap = flatArray.length - 1;
    rowCount = (flatArray[ap] & 255) + ((flatArray[ap - 1] & 255) << 8);
    ap -= 2;
    column = new ColumnQuerySupport[table.getColumnCount()];
    int columnEnd = flatArray.length - 2 - 3 * column.length;
    for (int i = column.length; --i > 0; ap -= 3) {
      column[i] = new LightColumn(columnEnd);
      columnEnd = (flatArray[ap] & 255)
                  + ((flatArray[ap - 1] & 255) << 8)
                  + ((flatArray[ap - 2] & 255) << 16);
    }
    if (column.length > 0) {
      column[0] = new LightColumn(columnEnd);
    }
  }

  @Override
  public Iterator<Host<RowVisitor>> iterator() {
    final IntIterator intIterator = IntIterators.fromTo(0, rowCount);
    final RecordHost recordHost = new RecordHost();
    return new Iterator<Host<RowVisitor>>() {
      @Override
      public boolean hasNext() {
        return intIterator.hasNext();
      }

      @Override
      public Host<RowVisitor> next() {
        recordHost.row = intIterator.nextInt();
        return recordHost;
      }

      @Override
      public void remove() {
        intIterator.remove();
      }
    };
  }

  @Override
  public Iterable<Host<RowVisitor>> getOrderedRange(
      int c, byte[] from, byte[] to, boolean asc) throws IOException {
    ColumnQuerySupport columnC = column[c];
    int start = from == null ? 0 : columnC.getRank(from);
    int end = to == null ? rowCount : columnC.getRank(to);
    return () -> {
      IntIterator intIterator = asc ? IntIterators.fromTo(start, end) : backwards(start, end);
      RecordHost host = new RecordHost();
      return new Iterator<Host<RowVisitor>>() {
        @Override
        public boolean hasNext() {
          return intIterator.hasNext();
        }

        @Override
        public Host<RowVisitor> next() {
          host.row = columnC.getRowAtRank(intIterator.nextInt());
          return host;
        }

        @Override
        public void remove() {
          intIterator.remove();
        }
      };
    };
  }

  private static  AbstractIntIterator backwards(final int start, final int end) {
    return new AbstractIntIterator() {
      int i = end;
      @Override
      public boolean hasNext() {
        return i > start;
      }

      @Override
      public int nextInt() {
        return --i;
      }
    };
  }

  /**
   * Manages values and ordered index of a string column. This is a flyweight implementation that
   * only stores the end of the column as an offset in {@link #flatArray}.
   * <pre>
   * Value {
   *   varInt length;
   *   byte[length] data;
   * }
   * Column {
   *   Value[row_count] value;
   *   int20[row_count] value_start;
   *   int16[row_count] index;  // value[index[i]] <= value[index[j]] if i <= j
   * }
   * </pre>
   */
  private class LightColumn implements ColumnQuerySupport {

    private final int end;

    private LightColumn(int end) {
      this.end = end;
    }

    /**
     * Searches index using binary search. Returns rank of the search key -- the count of rows
     * that have a strictly smaller value in the bound column.
     */
    @Override
    public int getRank(byte[] rhs) {
      int low = 0; // i >= low || value[i] < rhs
      int high = rowCount - 1; // i <= high || value[i] >= rhs
      CompareVisitor compareVisitor = new CompareVisitor(rhs);
      while (low <= high) {
        int mid = (low + high) >>> 1;
        acceptForRow(getRowAtRank(mid), compareVisitor);
        if (compareVisitor.diff < 0) {
          low = mid + 1;
        } else {
          high = mid - 1;
        }
      }
      return low;
    }

    @Override
    public int getRowAtRank(int rank) {
      int ap = end - 2 * (rowCount - rank);
      return ((flatArray[ap] & 255) << 8) + (flatArray[ap + 1] & 255);
    }

    @Override
    public void acceptForRow(int row, TypedCellVisitor visitor) {
      int ap = end - 2 * rowCount - 5 * ((rowCount + 1) / 2 - row / 2);
      int valueStart = (row & 1) == 0 ? evenInt20(ap) : oddInt20(ap);

      // Decode varInt length
      int length = 0;
      int tmp;
      do {
        tmp = flatArray[valueStart++] & 255;
        length = (length << 7) + (tmp & 127);
      } while ((tmp & 128) != 0);

      visitor.visitString(flatArray, valueStart, length);
    }

    private int evenInt20(int ap) {
      return ((flatArray[ap] & 255) << 12) + ((flatArray[ap + 1] & 255) << 4)
             + ((flatArray[ap + 2] & 240) >> 4);
    }

    private int oddInt20(int ap) {
      return ((flatArray[ap + 2] & 15) << 16) + ((flatArray[ap + 3] & 255) << 8)
             + (flatArray[ap + 4] & 255);
    }

    @Override
    public void appendDebugString(StringBuilder buf) {
      int ap = end - (2 * rowCount) - 5 * ((rowCount + 1) / 2);
      for (int i = 0; i < rowCount; i += 2, ap += 5) {
        buf.append("value_start: ")
            .append(((flatArray[ap] & 255) << 12)
                    + ((flatArray[ap + 1] & 255) << 4)
                    + ((flatArray[ap + 2] & 240) >> 4))
            .append('\n');
        buf.append("value_start: ")
            .append(((flatArray[ap + 2] & 15) << 16)
                    + ((flatArray[ap + 3] & 255) << 8)
                    + (flatArray[ap + 4] & 255))
            .append('\n');
      }
    }

    private class CompareVisitor extends TypedCellVisitor {

      private final byte[] rhs;
      int diff;

      public CompareVisitor(byte[] rhs) {
        this.rhs = rhs;
      }

      public void visitString(byte[] buf, int offset, int len) {
        diff = FilterQueryPlan.compare(buf, offset, len, rhs);
      }
    }
  }

  /**
   * Implements {@code Host<RowVisitor>} for the currently assigned record.
   */
  private final class RecordHost implements Host<RowVisitor> {
    int c;
    int row;
    private Host<TypedCellVisitor> cell = visitor -> column[c].acceptForRow(row, visitor);

    public void accept(RowVisitor visitor) {
      for (c = 0; c < column.length; c++) {
        visitor.visit(c, Type.STRING, cell);
      }
    }
  }

  StringBuilder appendDebugString(StringBuilder buf) {
    buf.append("version: ").append(((flatArray[0] & 255) << 8) + (flatArray[1] & 255)).append("\n");
    for (ColumnQuerySupport c : column) {
      c.appendDebugString(buf);
    }
    for (int ap = flatArray.length - 2 - 3 * column.length; ap < flatArray.length - 2; ap += 3) {
      buf.append("column_start: ").append(
          ((flatArray[ap] & 255) << 16)
          + ((flatArray[ap + 1] & 255) << 8)
          + (flatArray[ap + 2] & 255))
          .append('\n');
    }
    int p = flatArray.length - 2;
    buf.append("row_count: ")
        .append(((flatArray[p] & 255) << 8) + (flatArray[p + 1] & 255))
        .append('\n');
    return buf;
  }

  public String toDebugString() {
    return appendDebugString(new StringBuilder()).toString();
  }
}
